package com.lanier.dydownloader

import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.GlobalScope
import kotlinx.coroutines.awaitCancellation
import kotlinx.coroutines.cancel
import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.channels.awaitClose
import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.MutableSharedFlow
import kotlinx.coroutines.flow.MutableStateFlow
import kotlinx.coroutines.flow.asFlow
import kotlinx.coroutines.flow.callbackFlow
import kotlinx.coroutines.flow.channelFlow
import kotlinx.coroutines.flow.collect
import kotlinx.coroutines.flow.collectLatest
import kotlinx.coroutines.flow.conflate
import kotlinx.coroutines.flow.consumeAsFlow
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.flow.flowOn
import kotlinx.coroutines.flow.map
import kotlinx.coroutines.flow.merge
import kotlinx.coroutines.flow.receiveAsFlow
import kotlinx.coroutines.joinAll
import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking
import org.junit.Test
import kotlin.concurrent.thread
import kotlin.random.Random

// https://blog.csdn.net/m0_70748458/article/details/130286265
class TestFlow {

    @Test
    fun `test flow`() {
        runBlocking {
            //构造flow
            val flow = flow {
                //下游
                emit("hello world ${Thread.currentThread()}")
            }
            //收集flow
            flow.collect {
                //下游
                println("collect:$it ${Thread.currentThread()}")
            }
        }
    }

    //提供数据的模块
    class StudentInfo {
        fun getInfoFlow(): Flow<String> {
            return flow {
                //假装构造数据
                Thread.sleep(2000)
                emit("name=fish age=18")
            }
        }
    }

    //消费数据的模块
    @Test
    fun test1() {
        runBlocking {
            val flow = StudentInfo().getInfoFlow()
            flow.collect {
                println("studentInfo:$it")
            }
        }
    }

    @Test
    fun `test2 channel`() = runBlocking {
        //提前建立通道/管道
        val channel = Channel<String>()
        GlobalScope.launch(Dispatchers.Default) {
            //上游放数据（放水）
            delay(200)
            val data = "放水啦"
            println("上游:data=$data ${Thread.currentThread()}")
            channel.send(data)
        }

        GlobalScope.launch {
            val data = channel.receive()
            println("下游收到=$data ${Thread.currentThread()}")
        }


        delay(1500)
//        channel.close()
        println("作用域 done")
    }

    @Test
    fun `test6 channelFlow`() = runBlocking {
        //构造flow
        val channelFlow = channelFlow<String> {
            send("hello world emit 线程: ${Thread.currentThread()}")
        }
        channelFlow.collect {
            println("下游收到=$it collect 线程: ${Thread.currentThread()}")
        }
    }

    fun getName(callback: NetResult<String>) {
        thread {
            //假装从网络获取
            Thread.sleep(2000)
            callback.onSuc("I'm fish")
        }
    }

    interface NetResult<T> {
        fun onSuc(t: T)
        fun onFail(err: String)
    }

    @Test
    fun test7() {
        runBlocking {
            //构造flow
            val channelFlow = channelFlow {
                getName(object : NetResult<String> {
                    override fun onSuc(t: String) {
                        println("begin send")
                        trySend("hello world emit 线程: ${Thread.currentThread()}")
                        println("stop send")
                        //关闭channel，触发awaitClose闭包执行
                        close()
                    }

                    override fun onFail(err: String) {
                    }
                })

                //挂起函数
                awaitClose {
                    //走到此，channel关闭
                    println("awaitClose")
                }
            }

            channelFlow.collect {
                println("下游收到=$it collect 线程: ${Thread.currentThread()}")
            }
        }
    }

    @Test
    fun `test callbackFlow`() {
        runBlocking {
            //构造flow
            val channelFlow = callbackFlow {
                getName(object : NetResult<String> {
                    override fun onSuc(t: String) {
                        println("begin send")
                        trySend("hello world emit 线程: ${Thread.currentThread()}")
                        println("stop send")
                        //关闭channel，触发awaitClose闭包执行
                        close()
                    }

                    override fun onFail(err: String) {
                    }
                })

                //挂起函数
                awaitClose {
                    //走到此，channel关闭
                    println("awaitClose")
                }
            }

            channelFlow.collect {
                println("下游收到=$it collect 线程: ${Thread.currentThread()}")
            }
        }
    }

    @Test
    fun `test sharedFlow`() = runBlocking<Unit> {
        val sharedFlow = MutableSharedFlow<Int>()
        var count = 1

//        launch(Dispatchers.IO) {
//            repeat(10) {
//                val v = count + it
//                println("发送：$v, ${Thread.currentThread().name}")
//                sharedFlow.emit(v)
//                println("发送：$v done, ${Thread.currentThread().name}")
//            }
//        }

        repeat(100) {
            launch(Dispatchers.Default) {
                val v = it
                println("发送：$v, ${Thread.currentThread().name}")
                sharedFlow.emit(v)
                println("发送：$v done, ${Thread.currentThread().name}")
            }
        }

        launch {
            sharedFlow.collect {
//                count = it
                println("接收：$it, ${Thread.currentThread().name}")
            }
        }

    }

    @Test
    fun `test stateFlow`() = runBlocking<Unit> {
        val stateFlow = MutableStateFlow(0)

        repeat(1000) {
            launch(Dispatchers.Default) {
                val v = it
                println("发送：$v, ${Thread.currentThread().name}")
                stateFlow.emit(v)
                println("发送：$v done, ${Thread.currentThread().name}")
            }
        }

        launch {
            stateFlow.collect {
                println("接收：$it, ${Thread.currentThread().name}")
            }
        }

    }

    @Test
    fun `test channel 2 Flow`() = runBlocking<Unit> {
        val channel = Channel<Int>()
        val flow = channel.consumeAsFlow()

//        repeat(1000) {
//            launch(Dispatchers.Default) {
//                val v = it
//                println("发送：$v, ${Thread.currentThread().name}")
//                channel.send(v)
//                println("发送：$v done, ${Thread.currentThread().name}")
//            }
//        }

//        repeat(1000) {
//            launch {
//                delay(Random.nextLong(100))
//                val v = it
//                println("发送：$v, ${Thread.currentThread().name}")
//                channel.send(v)
//                println("发送：$v done, ${Thread.currentThread().name}")
//            }
//        }

        launch {
            repeat(1000) {
                delay(100)
                val v = it
                println("发送：$v, ${Thread.currentThread().name}")
                channel.send(v)
                println("发送：$v done, ${Thread.currentThread().name}")
            }
        }

        launch {
            flow.conflate().collect {
                println("接收：$it, ${Thread.currentThread().name}")
            }
        }

    }

    @Test
    fun `test flows`() = runBlocking<Unit> {
        val flow1 = flow<Int> {
            repeat(3) {
                emit(it)
            }
        }

        val flow2 = (4..6).asFlow()

        launch {
            merge(flow1, flow2).collect {
//                count = it
                println("接收：$it, ${Thread.currentThread().name}")
            }
        }

    }

}